package com.shanjupay.test.rocketmq.message;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.shanjupay.test.rocketmq.model.OrderExt;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.Charset;

/**
 * @className: ProducerSimple
 * @Description: TODO
 * @author: Huangjianyun
 * @date: 2021/10/26 14:40
 */
@Component
public class ProducerSimple {
    //自动注入rocketMQTemplate模板用来操作RocketMQ
    @Resource
    private RocketMQTemplate rocketMQTemplate;


    //发送同步消息方法
    //同步消息：消息生产者向broker发送消息，同步等待braker的API返回，直到broker返回结果
    public void  sendSyncMsg(String topic,String msg){
        //调用模板类的syncSend(主题，msg) broker的topic主体发送消息msg
        rocketMQTemplate.syncSend(topic,msg);
    }

    //发送异步消息
    //异步消息，生产者发送消息，返回发送成功或发送异常的的结果给我们消息生产者，另起一个线程进行做这个回调方法结果返回
    public void sendAsyncMsg(String topic,String msg){
        rocketMQTemplate.asyncSend(topic, msg, new SendCallback() {
            //消息发送成功的调用方法
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult);
            }

            //消息发送一场调用方法
            @Override
            public void onException(Throwable throwable) {
                System.out.println(throwable.getMessage());
            }
        });
    }
    //同步发送自定义的消息格式
    //实现原理：自定义的消息格式对象进行转化成json串，进行发送消息
    //源码：采用的同步发送消息的方式，有返回值

    public void sendMsgByJson(String topic, OrderExt orderExt){
        rocketMQTemplate.convertAndSend(topic,orderExt);
    }
    //同步发送延时消息
    public void sendMsgByJsonDelay(String topic,OrderExt orderExt){
        //1构建消息体，消息体所有spring的提供messaging类型
        Message<OrderExt> message = MessageBuilder.withPayload(orderExt).build();
        //2使用模板进行同步发送延迟消息 timeout:发送消息的超时时间 设置1s（单位ms） 级别设置3 对应10s
        rocketMQTemplate.syncSend(topic,message,1000,3);
    }

    //异步发送延迟消息
    public void sendAsynMsgByJsonDelay(String topic,OrderExt orderExt) throws JsonProcessingException, RemotingException, MQClientException, InterruptedException {
        //1消息内容进行转化成json
        String s = rocketMQTemplate.getObjectMapper().writeValueAsString(orderExt);
        //2构架rocketMQ提供消息体，进行消息自定义的封装
        org.apache.rocketmq.common.message.Message message =
                new org.apache.rocketmq.common.message.Message(topic, s.getBytes(Charset.forName("utf-8")));
        //3设置延迟等级3
        message.setDelayTimeLevel(3);
        //异步发送延迟的等级消息
        rocketMQTemplate.getProducer().send(message, new SendCallback() {
            //发送消息成功的回调消息
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult);
            }
            //发送消息异常的回调消息
            @Override
            public void onException(Throwable throwable) {
                System.out.println(throwable.getMessage());
            }
        });
        System.out.printf("send  msg : %s",orderExt);
    }


}
